package cn.dfun.sample.flink.wc
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._ // 包含StreamExecutionEnvironment

// 流处理word count
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度 任务按照key的hash分配
    // 开发环境默认并行度为cpu核心数
    env.setParallelism(2)

    // 从命令提取参数
    val paramTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = paramTool.get("host")
    val port: Int = paramTool.getInt("port")
    // 接收socket文本流
//    val inputDataStream = env.socketTextStream("node-01", 7777)
    val inputDataStream = env.socketTextStream(host, port)
    // 转换处理
    val resultDataStream = inputDataStream
      .flatMap(_.split(" "))
      // .setParallelism(3) // 每个算子都可以设置并行度,一般不用
      .map((_, 1))
      .keyBy(0)
      .sum(1)
    resultDataStream.print().setParallelism(1)
    // 启动任务执行
    env.execute("stream word count")
  }
}
